import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class TestAPI3 {
    Connection conn;
    Admin admin;
    TableName dianxinTT;
    @Before
    public void createConn() throws IOException {
        Configuration conf = HBaseConfiguration.create ( );
        conf.set ("hbase.zookeeper.quorum","master:2181,node1:2181,node2:2181");
         conn = ConnectionFactory.createConnection (conf);
         admin = conn.getAdmin ( );
        dianxinTT = TableName.valueOf ("dianxin");
    }
    @Test
    public void createDianxinTable() throws IOException {
        if (!admin.tableExists (dianxinTT)){
            admin.createTable (new HTableDescriptor (dianxinTT).addFamily (new HColumnDescriptor ("info").setMaxVersions (4)));
        }else{
            System.out.println ("表已存在" );
        }
    }
    @Test
    public void putall() throws IOException {
        Table table = conn.getTable (dianxinTT);
        BufferedReader br = new BufferedReader (new FileReader ("H:\\soft\\IDEA1.8\\bd13\\Hbase\\src\\DIANXIN.csv"));
        String line;
        int batchSize=1000;
        ArrayList<Put> puts = new ArrayList<Put> ( );
        while((line = br.readLine()) != null) {
            String[] split = line.split (",");
            String mdn = split[0];
            String start = split[1];
            String lg = split[4];
            String lat = split[5];
            Put put = new Put (mdn.getBytes ( ));
            put.addColumn ("info".getBytes ( ), "lg".getBytes ( ), Long.parseLong (start), lg.getBytes ( ));
            put.addColumn ("info".getBytes ( ), "lat".getBytes ( ), Long.parseLong (start), lat.getBytes ( ));
            puts.add (put);
            if (puts.size ( ) == batchSize) {
                table.put (puts);
                puts.clear ( );
            }
        }
            if (!puts.isEmpty ()){
                table.put (puts);
            }
    }
    @After
    public void close() throws IOException {
        admin.close ();
        conn.close ();

    }
}
